/*
 *  ------------------------------------------------------------------
 *  Copyright © 2017 Hangzhou DtDream Technology Co.,Lt d. All rights reserved.
 *  ------------------------------------------------------------------
 *              Product: EMR
 *    Module Name: DataTrain
 *      Date Created: 2017-10-23
 *         Description:
 *  ------------------------------------------------------------------
 *  Modification History
 *  DATE            Name           Description
 *  ------------------------------------------------------------------
 *  2017-10-23     XZ 209
 *  ------------------------------------------------------------------
 */

package org.xukai.remoting.sdk.rpc.job;


import com.dtdream.emr.transmission.data.EmrSchema;
import com.dtdream.emr.transmission.exception.TransmissionException;
import com.dtdream.emr.transmission.io.Constant;

import java.util.Arrays;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import org.xukai.remoting.sdk.rpc.transfer.NettyConfig;
import org.xukai.remoting.sdk.rpc.transfer.TransferClient;

import static com.dtdream.emr.transmission.exception.TransmissionException.Error.EXCEPTION_SDK_NETWORK;

public abstract class Job implements AutoCloseable {
    private String projectName;
    private String tableName;
    private String partition; // format: "a=1,b=2"
    private EmrSchema schema;
    private UUID jobId;

    private String serverUrls;
    private String accessKey;
    private String keySecret;

    public TransferClient transferClient;
    private boolean closed = false;

    private static ConcurrentHashMap<UUID, Job> jobs = new ConcurrentHashMap<>();

    /**
     * <p>创建job时需要设置的属性有：<br>
     * "address"，服务地址，多节点之间逗号分隔；<br>
     * "accesskey", 用户Access key；<br>
     * "keysecret"，用户Access Key对应的Key secret；<br>
     * "projectname"，project名称；<br>
     * "tablename", 表名；<br>
     * "partition"，分区表的分区，各字段之间逗号分隔</p>
     *
     * @param properties
     * @throws TransmissionException
     */
    public Job(Properties properties) throws TransmissionException {
        Job.validateJobProperties(properties);
        this.projectName = properties.getProperty(Constant.PROJECTNAME);
        this.tableName = properties.getProperty(Constant.TABLENAME);
        this.partition = properties.getProperty(Constant.PARTITION);

        this.serverUrls = properties.getProperty(Constant.URL);
        this.accessKey = properties.getProperty(Constant.ACCESSKEY);
        this.keySecret = properties.getProperty(Constant.KEYSECRET);
        NettyConfig nettyConfig = new NettyConfig();
        nettyConfig.setEventLoopThreads(Integer.parseInt(properties.getProperty("pool.size", "4")));
        // default high_watermark is 4MB,low_watermark is half of high_watermark.
        nettyConfig.setWaterMark(Integer.parseInt(properties.getProperty("watermark", "4194304")));
        nettyConfig.assertValid();
        try {
            transferClient = new TransferClient(this.serverUrls, accessKey, keySecret, nettyConfig);
            transferClient.init();
            jobId = createJobInRemote();
            schema = transferClient.getEmrSchema(jobId);
        } catch (TransmissionException e) {
              try {
                close();
            } catch (Exception ie) {
                e.addSuppressed(ie);
            } finally {
                throw e;
            }
        }

        Job.addJob(this);
    }

    public static void addJob(Job job) {
        jobs.put(job.getJobId(), job);
    }

    public static Job removeJob(UUID jobId) {
        return jobs.remove(jobId);
    }

    public static Job getJob(UUID jobId) {
        return jobs.get(jobId);
    }

    public static void validateJobProperties(Properties properties) {
        for (String key : Arrays.asList(Constant.PROJECTNAME, Constant.TABLENAME, Constant.URL, Constant.ACCESSKEY,
                Constant.KEYSECRET)) {
            if (properties.getProperty(key) == null) {
                throw new IllegalArgumentException(key + " must not be null");
            }
        }
    }

    protected abstract UUID createJobInRemote() throws TransmissionException;

    public void close() throws TransmissionException {
        if (closed) {
            return;
        } else {
            closed = true;
        }

        try {
            if (jobId != null) {
                this.transferClient.closeJob(jobId);
                Job.removeJob(jobId);
            }
        } finally {
            this.transferClient.close();
        }
    }

    public String getProjectName() {
        return projectName;
    }

    public void setProjectName(String projectName) {
        this.projectName = projectName;
    }

    public String getTableName() {
        return tableName;
    }

    public void setTableName(String tableName) {
        this.tableName = tableName;
    }

    public String getPartition() {
        return partition;
    }

    public void setPartition(String partition) {
        this.partition = partition;
    }

    public UUID getJobId() {
        return jobId;
    }

    public void setJobId(UUID jobId) {
        this.jobId = jobId;
    }

    public EmrSchema getSchema() {
        return schema;
    }

    public void setSchema(EmrSchema schema) {
        this.schema = schema;
    }
}


